package com.jd.rapid.rpc.client;

import com.jd.rapid.rpc.codec.RpcRequest;
import com.jd.rapid.rpc.codec.RpcResponse;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

import java.awt.*;
import java.net.SocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;


/**
 * 实际的业务处理器
 */
public class RpcClientHandler extends SimpleChannelInboundHandler<RpcResponse> {

    private Channel channel;

    private SocketAddress remotePeer;

    private Map<String/**requestId**/,RpcFuture> pendingRpcTable=new ConcurrentHashMap<>();

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        super.channelRegistered(ctx);
        //注册的时候，可以拿到channel
        this.channel=ctx.channel();
    }


    /**
     * 通道激活的时候触发此方法
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        //激活的时候，可以拿到address
        this.remotePeer=this.channel.remoteAddress();
    }


    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, RpcResponse rpcResponse) throws Exception {
        //接受响应
        String requestId=rpcResponse.getRequestId();
          RpcFuture rpcFuture=pendingRpcTable.get(requestId);
          if(rpcFuture!=null){
              pendingRpcTable.remove(requestId);
              rpcFuture.done(rpcResponse);
          }
    }

    public SocketAddress getRemotePeer() {
        return this.remotePeer;
    }

    /**
     * netty提供一种主动关闭连接的方式，发送一个Unpooled.EMPTY_BUFFER
     * 这样ChannelFutureListener.CLOSE的事件就会监听到并关闭连接
     */
    public void close() {
        //触发事件，空的buffer，ChannelFutureListener.CLOSE
        channel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
    }


    /**
     * 异步发送请求方法
     * @param request
     * @return
     */
    public RpcFuture sendRequest(RpcRequest request){
        RpcFuture rpcFuture=new RpcFuture(request);
        //异步放入缓冲
        pendingRpcTable.put(request.getRequestId(),rpcFuture);
        channel.writeAndFlush(request);
        return rpcFuture;
    }

    public Channel getChannel() {
        return channel;
    }
}
